package com.xiashitech.agent.grpc;

import com.xiashitech.agent.config.StartConfig;
import com.xiashitech.agent.utils.JsonUtil;
import com.xiashitech.interfaces.agent.dto.AgentConfigDTO;
import com.xiashitech.interfaces.agent.dto.SystemInfoDTO;
import com.xiashitech.interfaces.apiservice.dto.AuthFactRequest;
import com.xiashitech.interfaces.grpc.*;
import io.grpc.Deadline;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import io.opentelemetry.javaagent.bootstrap.internal.InstrumentationConfig;

import java.util.Iterator;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class XsConfigClient {
    private final static XsConfigClient client = new XsConfigClient();

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    private volatile boolean isError = false;
    private boolean isSecheduleStarted = false;
    public static XsConfigClient getClient() {
        return client;
    }


    public String blockingSyncConfig() {
        String confStr = null;
        try {
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", blockingSyncConfig_E1 enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
            ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", blockingSyncConfig_E2, now channel identityHashCode: " + System.identityHashCode(channelRef));

            RPCConfigServiceGrpc.RPCConfigServiceBlockingStub blockingStub = RPCConfigServiceGrpc.newBlockingStub(channelRef);
            Deadline dl = Deadline.after(60, TimeUnit.SECONDS);
            blockingStub = blockingStub.withDeadline(dl);
            // 服务端目前不需要请求参数
            String microServiceId = InstrumentationConfig.get().getString("otel.service.name");
            SystemInfoDTO siDto = SystemInfoDTO.builder().id(StartConfig.getParamValueFromKey(StartConfig.systemId)).microServiceId(microServiceId).build();
            RPCConfigRequest request = RPCConfigRequest.newBuilder().setSysinfoJson(JsonUtil.convertObjectToString(siDto)).setType("blocking").build();
            Iterator<RPCConfigResponse> responseIterator = blockingStub.grpcSyncConfig(request);
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", blockingSyncConfig_A, deadline 60s, start hasNext: ");
            if (responseIterator.hasNext()) {
                RPCConfigResponse response = responseIterator.next();
                confStr = response.getAgentConfigDTOJson();
                System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", blockingSyncConfig_B, hasNext, get the response");
            }
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", blockingSyncConfig_C, finished get the config: " + confStr);
        } catch (Exception e) {
            StartConfig.throwableStackPrinter(e);
        }
        return confStr;
    }

    public synchronized void grpcSyncConfig() { // 实际业务不会并发调用 grpcSyncConfig，加 synchronized 避免错误的并发调用
        System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig, now channel identityHashCode: "+ System.identityHashCode(channelRef));
        StreamObserver<RPCConfigResponse> configResponse = new StreamObserver<RPCConfigResponse>() {
            @Override
            public void onNext(RPCConfigResponse value) {
                String configJson = value.getAgentConfigDTOJson();
                System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig receive the config response: " +configJson);
                if("CONFIGHEARTBEAT".equals(configJson)) {
                    System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig received the pushed heartbeat: " +configJson);
                } else if(configJson != null && !configJson.isEmpty() && JsonUtil.isJson(configJson)) {
                    AgentConfigDTO.setConfig(JsonUtil.convertStringToObject(configJson, AgentConfigDTO.class));
                } else {
                    System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig the pulled config str is not correct: " +configJson);
                }
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig channel has broken, onError: T" + System.nanoTime() +", AAgrpcSyncConfigAA: " + t.getMessage());
                isError = true;
            }

            @Override
            public void onCompleted() {
                System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig channel completed: onCompleted");
            }
        };
        // 服务端目前不需要请求参数
        String microServiceId = InstrumentationConfig.get().getString("otel.service.name");
        SystemInfoDTO siDto = SystemInfoDTO.builder().id(StartConfig.getParamValueFromKey(StartConfig.systemId)).microServiceId(microServiceId).build();
        RPCConfigRequest request = RPCConfigRequest.newBuilder().setSysinfoJson(JsonUtil.convertObjectToString(siDto)).setType("nonblocking").build();
        RPCConfigServiceGrpc.RPCConfigServiceStub stub = RPCConfigServiceGrpc.newStub(channelRef);
        stub.grpcSyncConfig(request, configResponse);
        System.out.println(Thread.currentThread().getId() + ", grpcSyncConfig, method finish");
        // 启动定时器，开始自动重连
        if(!isSecheduleStarted) {
            isSecheduleStarted = true;
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                if (isError) {
                    isError = false;
                    grpcSyncConfig();
                }
            }, 5, 10, TimeUnit.SECONDS);
        }
    }
    public String grpcInitSystemInfo(AuthFactRequest authFactRequest) {
        System.out.println(Thread.currentThread().getId() + ", grpcInitSystemInfo enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", grpcInitSystemInfo, now channel identityHashCode: "+ System.identityHashCode(channelRef));

        RPCConfigServiceGrpc.RPCConfigServiceBlockingStub blockingStub = RPCConfigServiceGrpc.newBlockingStub(channelRef);
        Deadline dl = Deadline.after(20, TimeUnit.SECONDS);
        blockingStub = blockingStub.withDeadline(dl);
        RPCSystemInfoRequest request = RPCSystemInfoRequest.newBuilder().setSystemInfoRequestJson(JsonUtil.convertObjectToString(authFactRequest)).build();
        RPCSystemInfoResponse response = blockingStub.grpcInitSystemInfo(request);
        return response.getSystemInfoResponseJson();
    }
}
